package com.peng.sparktest.sparkstreaming

import org.apache.spark.streaming.dstream.{DStream, MapWithStateDStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Duration, State, StateSpec, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object StreamTest03_DStreamApiTest02 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("dstreamApiTest")

    val context = new SparkContext(conf)

    context.setLogLevel("ERROR")
    context.setCheckpointDir("file:///test_file/ckp")

    val sc = new StreamingContext(context, Duration(1000)) //Duration和Second的区别是，一个是毫秒，一个是秒

    val dStream: ReceiverInputDStream[String] = sc.socketTextStream("localhost", 8888)

    val mapData: DStream[(String, Int)] = dStream.map(_.split(" ")).map(item => (item(0), 1))

    //对拉取的数据进行“全量有状态计算”，  让每次的聚合操作可以基于历史记录的数据进行合计等操作
    //这里不难想到，实现的原理就是persist和checkpoint
    //每一时刻拉取批次的数据计算后会被持久化起来，后续的新批数据就可以基于上一次的计算后持久化数据进行继续的计算
    //（不过这种做法基本不怎么用，因为它时简单粗暴的将同key下的全量一批数据去执行此方法，会有内存溢出的风险！）
    //    val res: DStream[(String, Int)] = mapData.updateStateByKey((newV: Seq[Int], oldV: Option[Int]) => {
    //      println(s"newV:${newV}  oldV:${oldV}")
    //      val count: Int = newV.count((_: Int) > 0)
    //      val oldCount: Int = oldV.getOrElse(0)
    //      Some(count + oldCount)
    //    })
    //    res.print()

    //这种才是常用的： 一条一条的去执行方法
    val res: MapWithStateDStream[String, Int, Int, (String, Int)] = mapData.mapWithState(StateSpec.function(
      (key: String, nv: Option[Int], ov: State[Int]) => {
        println(s"key:$key , nv:${nv} , ov:${ov.getOption()}")
        (key, nv.getOrElse(0) + ov.getOption().getOrElse(0))
      }
    ))
    res.print()









    //调优：如果是窗口滑动的数据，有一种场景是 每秒进行刷新历史5分钟内的数据批次，这个时候，怎么做比较好？
    //第一种想到的肯定就是，每秒去对历史5分钟全量数据都进行计算一次，得出结果
    //但会存在问题：5分钟的全量数据计算，如果数据比较多，每秒都能够计算完吗？ 是不是会造成由于等待处理导致的batch累积
    //所以一种优化方案是，可以减去最早1秒的数据， 然后加上新1秒的数据  ，这样只需要进行相对很少的数据就能完成计算
    //这种方法，特别是在窗口数据比较多的情况下，效果比较显著。。 当然，也要考虑这样做会不会导致逻辑与原有逻辑不一样
    //当然， “减去最早1秒的数据， 然后加上新1秒的数据”要有个前提，就是持有历史的数据状态记录，基于这个数据进行操作
    //因此，依旧需要使用到checkpoint的持久化数据

    //    val res: DStream[(String, Int)] = mapData.reduceByKeyAndWindow(
    //      (oldV: Int, newV: Int) => {
    //        println(s"reduceFunc: oldV:$oldV ,newV:$newV")
    //        oldV + newV
    //      },
    //      (oldV: Int, outV: Int) => {
    //        println(s"invReduceFunc:oldV:$oldV,outV:$outV")
    //        //oldV是历史的数据
    //        //outV是窗口滚动后，被挤出去的数据
    //        oldV - outV
    //      }
    //      , Duration(6000),
    //      Duration(2000)
    //    )
    //    res.print()

    //   hello 1
    // hello 1
    // world 1

    //   hello 1
    // hello 1
    // world 1

    // hello 2
    // world 1

    // hello 2
    // world 1


    sc.start()
    sc.awaitTermination()
  }

}
